package com.niit.init

import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object ParseInfo {
  val MONGODB_RATING_COLLECTION = "Rating"
//  val RATING_DATA_PATH = "E:\\IdeaProjects\\recommendersys\\businessSys\\logs\\log4j2\\info.log"
  val RATING_DATA_PATH = "/training/businessSys/logs/log4j2/info.log"
  def main(args: Array[String]): Unit = {
    // 定义用到的配置参数
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://fooadmin:123456@127.0.0.1:27017/bigdata",
      "mongo.db" -> "bigdata"
    )
    // 创建一个SparkConf配置
    val sparkConf = new SparkConf().setAppName("ParseInfo").setMaster(config("spark.cores"))
    // 创建一个SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 在对DataFrame和Dataset进行操作许多操作都需要这个包进行支持
    import spark.implicits._
    // 将Product、Rating数据集加载进来
    val rating2RDD = spark.sparkContext.textFile(RATING_DATA_PATH)

    val rating2DF = rating2RDD.filter(_.contains("PRODUCT_RATING_PREFIX:")) // PRODUCT_RATING_PREFIX:73214636|62138|5.0|1603760681
      .map(item => { // PRODUCT_RATING_PREFIX:73214636|62138|5.0|1603760681
        val strings = item.split("PRODUCT_RATING_PREFIX:")(1).trim()
        val arrays = strings.split("\\|")
        Rating(arrays(0).toInt,arrays(1).toInt,arrays(2).toDouble,arrays(3).toInt)
      }).toDF()

    val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))
    val rating2Collection = mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    rating2Collection.dropCollection()

    rating2DF.write
      .option("uri",mongoConfig.uri)
      .option("collection",MONGODB_RATING_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    mongoClient.close()

    spark.stop()
  }
}
